package org.hbhk.hms.mq;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.hbhk.hms.entity.HmsTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;

public class MqSender {
	private static Logger log = LoggerFactory.getLogger(MqSender.class);

	public static SendResult sendMessage(HmsTopic hmsTopic) {

		try {
			init();

			String topic = hmsTopic.getTargetTopic();
			String key = hmsTopic.getMsgKey();
			String mqMessageStr = hmsTopic.getMsg();
			String tags = hmsTopic.getTags();
			if (StringUtils.isEmpty(tags)) {
				tags = "*";
			}
			String mqName = hmsTopic.getMqName();
			DefaultMQProducer template = MqProvider.get(mqName);
			Message msg = new Message(topic, tags, key, mqMessageStr.getBytes());
			SendResult sendResult = null;
			sendResult = template.send(msg);
			if (!(sendResult.getSendStatus().equals(SendStatus.SEND_OK))) {
				throw new RuntimeException("MQ send  failed:" + sendResult.getSendStatus().toString());
			}
			return sendResult;

		} catch (Exception e) {
			log.error("MQ send  failed", e);
			throw new RuntimeException("MQ send  failed", e);
		}
	}

	public static SendResult sendMessage(List<HmsTopic> hmsTopics) {
		try {
			init();
			List<Message> msgList = Lists.newArrayList();
			String mqName = null;
			for (HmsTopic hmsTopic : hmsTopics) {
				String topic = hmsTopic.getTargetTopic();
				String key = hmsTopic.getMsgKey();
				String mqMessageStr = hmsTopic.getMsg();
				String tags = hmsTopic.getTags();
				if (StringUtils.isEmpty(tags)) {
					tags = "*";
				}
				if (mqName == null) {
					mqName = hmsTopic.getMqName();
				}
				Message msg = new Message(topic, tags, key, mqMessageStr.getBytes());
				msgList.add(msg);
			}

			DefaultMQProducer template = MqProvider.get(mqName);
			SendResult sendResult = null;
			sendResult = template.send(msgList);
			if (!(sendResult.getSendStatus().equals(SendStatus.SEND_OK))) {
				throw new RuntimeException("MQ send  failed:" + sendResult.getSendStatus().toString());
			}
			return sendResult;

		} catch (Exception e) {
			log.error("MQ send  failed", e);
			throw new RuntimeException("MQ send  failed", e);
		}
	}

	public static SendResult sendMessage(List<HmsTopic> hmsTopics, String mqName, String topic) {
		try {
			init();

			List<Message> msgList = Lists.newArrayList();
			for (HmsTopic hmsTopic : hmsTopics) {
				String key = hmsTopic.getTargetTopic() + "_" + hmsTopic.getMsgKey();
				String mqMessageStr = JSON.toJSONString(hmsTopic);
				Message msg = new Message(topic, "*", key, mqMessageStr.getBytes());
				msgList.add(msg);
			}

			DefaultMQProducer template = MqProvider.get(mqName);

			SendResult sendResult = template.send(msgList);
			if (!(sendResult.getSendStatus().equals(SendStatus.SEND_OK))) {
				throw new RuntimeException("MQ send  failed:" + sendResult.getSendStatus().toString());
			}
			return sendResult;

		} catch (Exception e) {
			log.error("MQ send  failed", e);
			throw new RuntimeException("MQ send  failed", e);
		}
	}

	public static SendResult sendMessage(String key, String msgStr, String mqName, String topic) {
		try {
			init();
			Message msg = new Message(topic, "*", key, msgStr.getBytes());
			DefaultMQProducer template = MqProvider.get(mqName);
			SendResult sendResult = template.send(msg);
			if (!(sendResult.getSendStatus().equals(SendStatus.SEND_OK))) {
				throw new RuntimeException("MQ send  failed:" + sendResult.getSendStatus().toString());
			}
			return sendResult;

		} catch (Exception e) {
			log.error("MQ send  failed", e);
			throw new RuntimeException("MQ send  failed", e);
		}
	}

	private static void init() {
		while (MqProvider.start = false) {
			try {
				TimeUnit.MILLISECONDS.sleep(200);
				log.error("mq provider loading ...");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}
}
